package com.shujia.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {

    /**
      * 1、创建spark上下文环境
      * SparkContext是spark代码的入口
      *
      */

    //spark配置文件对象，指定spark 运行常用的配置
    val conf = new SparkConf()

    //任务名
    conf.setAppName("Demo1WordCount")

    //指定运行模式，local :本地运行
    conf.setMaster("local")

    //创建上下文对象
    val sc = new SparkContext(conf)

    /**
      * 1、读取文件,  读取hdfs中的文件
      *
      * RDD: 弹性的分布式数据集  -- 可以理解为一个分布式的List集合
      *
      */

    val linesRDD: RDD[String] = sc.textFile("data/words")


    /**
      * 2、统计单词的数量
      *
      * flatMap: 算子，返回一个新的RDD
      *
      */

    //1、将一行拆分成多行
    val wordsRDD: RDD[String] = linesRDD.flatMap((line: String) => line.split(","))


    /**
      * groupBy: 按照指定的key分组： --  在执行的底层会产生shuffle
      *
      * 迭代器和集合的区别
      * 1、迭代器只能遍历一次
      * 2、迭代器的数据没有完全在内存中，集合的数据完全保存在内存中
      *
      */
    //3、按照单词分组
    val groupByRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy((word: String) => word)


    //4、统计单词的数量
    val countRDD: RDD[(String, Int)] = groupByRDD.map {
      case (word: String, words: Iterable[String]) =>
        //单词的数量
        val count: Int = words.size

        //返回数据
        (word, count)
    }

    //5、整理数据

    val resultRDD: RDD[String] = countRDD.map {
      case (word: String, count: Int) =>
        s"$word\t$count"
    }

    /**
      * 3、保存数据
      *
      * 指定的路径是一个目录，里面文件的数量和reduce的数量有关（和mr一样）
      *
      */
    resultRDD.saveAsTextFile("data/word_count")


  }

}
